package de.preissler.insertjoin;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.RangeSlicesQuery;

public class Main
{
	public static String HOST;
	public static final String KEYSPACE_NAME = "Exchange";
	private static final String QUOTES = "Quotes";
	private static final int YEAR = 2011;
	private static final int MONTH = 2;
	private static final int DAY = 10;
	private static final String[] DAX_ISINS = {"DE000A1EWWW0", "DE0008404005", "DE000BASF111", "DE0005190003",
			"DE000BAY0017", "DE0005200000", "DE0008032004", "DE0007100000", "DE0005140008", "DE0005810055",
			"DE0005552004", "DE0005557508", "DE000ENAG999", "DE0005785802", "DE0005785604", "DE0006047004",
			"DE0006048432", "DE0006231004", "DE0007162000", "DE0006483001", "DE0008232125", "DE0005937007",
			"DE0006599905", "DE0007257503", "DE0008430026", "DE0007037129", "DE0007164600", "DE0007236101",
			"DE0007500001", "DE0007664039"};
	private static final String DAX_ISIN = "DE0008469008";
	
	public static void main(String[] args) throws Exception
	{
		//Cluster initialisieren
		Cluster cluster = getCluster();

		Keyspace keyspaceOperator = HFactory.createKeyspace(KEYSPACE_NAME, cluster);
		keyspaceOperator.setConsistencyLevelPolicy(new me.prettyprint.cassandra.model.AllOneConsistencyLevelPolicy());
		
		//Zufaellige Kurse erzeugen
		int numberOfDays = getNumberOfDays();
		Map<String, Map<String, Quote>> quotes = getQuotes(keyspaceOperator, numberOfDays);
		Map<String, Map<String, DAXParam>> params = getDAXParams(keyspaceOperator, numberOfDays);
		
		Mutator<String> mutator = HFactory.createMutator(keyspaceOperator, StringSerializer.get());
		
		InsertData insAndDel = getInsAndDel();
		
		//speichert die gemessene Zeit
		long[] time = new long[3];
		
		//das ganze 3 mal machen
		for(short i = 0; i < 3; i++)
		{
			long start = System.nanoTime();
			
			insAndDel.insert(quotes, params, keyspaceOperator, mutator);
			
			long end = System.nanoTime();
			
			time[i] = end - start;
			
			insAndDel.delete(quotes, params, keyspaceOperator, mutator);
		}
		
		//Hector sauber beenden
		cluster.getConnectionManager().shutdown();
		
		for(long i : time)
		{
			System.out.println(i + " ns");
		}
	}
	
	private static InsertData getInsAndDel() throws Exception
	{
		System.out.println("<1> LVT and RVT     <2> RVT     <3> Calc");
		int choice = Integer.valueOf(readLine());
		switch(choice)
		{
			case 1: return new InsertDataPreSaved();
			case 2: return new InsertDataJoin();
			case 3: return new InsertDataCalc();
			default: return null;
		}
	}

	/*
	 * Holt alle ISINs raus
	 */
	private static List<String> getKeys(Keyspace keyspace, String nameOfColumnFamily)
	{
		List<List<String>> listsOfISINs = new LinkedList<List<String>>();
		
		String start = null;
		String end = null;
		boolean firstLoop = true;
		//Max. werden nur 100 Schluessel zurueck gegeben, deswegen diese Schleife
		while(true)
		{
			List<String> isins = new LinkedList<String>();
			
	         RangeSlicesQuery<String, String, String> rangeSlicesQuery = 
	        	 HFactory.createRangeSlicesQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
		     rangeSlicesQuery.setColumnFamily(nameOfColumnFamily);
		     rangeSlicesQuery.setReturnKeysOnly();
		     rangeSlicesQuery.setKeys(start, end);
		     List<Row<String, String, String>> result = rangeSlicesQuery.execute().get().getList();
		     
		     for(int i = 0; i < result.size(); i++)
		     {
		    	 if((!firstLoop) && (i == 0))
		    	 {
		    		 
		    	 }
		    	 else
		    		 isins.add(result.get(i).getKey());
		    		 
		     }
		     
		     if(isins.isEmpty())
		    	 break;
		     else
		     {
		    	 start = isins.get(isins.size()-1);
		    	 listsOfISINs.add(isins);
		    	 firstLoop = false;
		     }
		}
		
		List<String> isins = new LinkedList<String>();
		for(List<String> subList : listsOfISINs)
		{
			isins.addAll(subList);
		}
	     
		return isins;
	}
	
	private static List<String> getDates(int number)
	{
		List<String> result = new LinkedList<String>();
		
		for(int i = DAY; i < number+DAY; i++)
		{
			result.add(YEAR + "-" + "0" + MONTH + "-" + i);
		}
		return result;
	}
	
	private static int getNumberOfDays() throws Exception
	{
		System.out.println("For how many days, courses should be calculated?");
		return Integer.valueOf(readLine());
	}
	
	private static Map<String, Map<String, DAXParam>> getDAXParams(Keyspace keyspace, int numberOfDays)
	{
		List<String> dates = getDates(numberOfDays);
		
		Map<String, Map<String, DAXParam>> result = new HashMap<String, Map<String, DAXParam>>(dates.size());
		for(String date : dates)
		{
			Map<String, DAXParam> subresult = new HashMap<String, DAXParam>(DAX_ISINS.length+1);
			for(String isin : DAX_ISINS)
			{
				double kt = 0.0;
				double qit = Math.random() * 10000000.0;
				double ffi = 1.0;
				double ci = 1.0;
				double pi0 = Math.random() * 100.0;
				double qi0 = Math.random() * 1000000.0;
				double base = 0.0;
				subresult.put(isin, new DAXParam(kt, qit, ffi, ci, pi0, qi0, base));
			}
			
			//DAX selbst einfuegen
			double kt = Math.random()*2;
			double qit = 0.0;
			double ffi = 0.0;
			double ci = 0.0;
			double pi0 = 0.0;
			double qi0 = 0.0;
			double base = 1000.0;
			subresult.put(DAX_ISIN, new DAXParam(kt, qit, ffi, ci, pi0, qi0, base));
			
			result.put(date, subresult);
		}
		
		return result;
	}
	
	/*
	 * Erstellt die zufaelligen Kurse
	 */
	private static Map<String, Map<String, Quote>> getQuotes(Keyspace keyspace, int numberOfDays) throws Exception
	{
		List<String> dates =  getDates(numberOfDays);
		List<String> isins = getKeys(keyspace, QUOTES);

		Map<String, Map<String, Quote>> result = new HashMap<String, Map<String, Quote>>(dates.size());
		for(String date : dates)
		{
			Map<String, Quote> subresult = new HashMap<String, Quote>(isins.size());
			for(String isin : isins)
			{
				double change = Math.random();
				double close = Math.random() * 100.0;
				double high = Math.random() * 100.0;
				double low = Math.random() * 100.0;
				double open = Math.random() * 100.0;
				double volume = Math.random() * 1000000.0;
				subresult.put(isin, new Quote(change, close, high, low, open, volume));
			}
			result.put(date, subresult);
		}
		
		return result;
	}
	
	private static String readLine() throws Exception
	{
		BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
		return in.readLine();
	}
	
	/*
	 * Erstellt das Cluster Objekt
	 */
	private static Cluster getCluster() throws Exception
	{
		String clusterName;
		String host;
		String port;
		
		System.out.println("In order to connect to cassandra we need some parameters...");
		
		System.out.print("clustername: ");
		clusterName = readLine();
		
		System.out.print("host: ");
		host = readLine();
		HOST = host;
		
		System.out.print("port: ");
		port = readLine();
		
		return HFactory.getOrCreateCluster(clusterName, host + ":" + port);
	}
}
